package com.heima.kafka.sample;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.Test;

import java.util.Properties;

/**
 * 生产者
 */
public class ProducerQuickStart {

    @Test
    public void testSendMsg() {
        // 添加kafka的配置信息
        Properties properties = new Properties();
        // 添加连接配置
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "81.68.105.177:9092");
        // 添加发送失败后的重试次数
        properties.put(ProducerConfig.RETRIES_CONFIG, 5);
        // 添加key的序列化器
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        // 添加value的序列化器
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");

        // 创建生产者核心对象
        Producer<String, String> producer = new KafkaProducer<String, String>(properties);

        for (int i = 0; i < 5; i++) {
            // 封装发送的消息
            ProducerRecord<String, String> producerRecord = new ProducerRecord<>("itcast_topic_input", "KStream", "lian xi shi chang liang nian ban");
            // 发送消息
            producer.send(producerRecord);

            System.out.println("消息发送成功!!!!");
        }

        // 释放资源
        producer.close();
    }


}
